#include "IOManager.hpp"
#include "../base/BaseLog.hpp"
#include "Fiber.hpp"
#include "Scheduler.hpp"
#include <cassert>
#include <cerrno>
#include <cstring>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <sys/epoll.h>
#include <cstddef>
#include <cstdint>
#include <functional>
#include <stdexcept>
#include <string>
#include <unistd.h>
#include <vector>

using namespace DBServer::coroutine;

IOManager::IOContext::EventContext& IOManager::IOContext::get_ctx(IOEvent ev) {
    switch (ev) {
        case IOEvent::READ:
            return read;
        case IOEvent::WRITE:
            return write;
        case IOEvent::HUP:
            return hup;
        default:
            error("invaild IOEvent(%d)\n", (int)ev);
    }
    throw std::invalid_argument("invaild argument of IOEvent");
}

void IOManager::IOContext::reset_ctx(IOManager::IOContext::EventContext &ctx) {
    ctx.cb = nullptr;
    ctx.fiber.reset();
    ctx.scher = nullptr;
}

void IOManager::IOContext::trigger_ev(IOEvent ev) {
    assert((int)ev & (int)events);

    // 获取对应事件的上下文并将其事件移除，
    // 因为每一个事件只触发一次
    auto &ctx = get_ctx(ev);
    events = ~(int)ev & (int)events;

    if (ctx.cb) {
        ctx.scher->schedule(&ctx.cb);
    } else {
        ctx.scher->schedule(&ctx.fiber);
    }
}

IOManager::IOManager(std::size_t threads, bool caller, const std::string &name)
: Scheduler(threads, caller, name)
{
    int epoll_fd = epoll_create(5000);
    if (epoll_fd < 0) {
        error("epoll_create fail(%d): %s\n", epoll_fd, strerror(errno));
        throw;
    }
    info("epoll_create epfd(%d)\n", epoll_fd);
    epfd = epoll_fd;
}

IOManager::~IOManager() {
    close(epfd);

    for (auto &entry : io_ctxs) 
    {
        delete entry.second;
    }
    io_ctxs.clear();
}

int IOManager::add_ev(int fd, IOEvent ev, std::function<void()> cb) {
    if (stopping()) {
      warn("io manager is stopping!\n");
      return -1;
    }
  
    IOContext * io_ctx;
    std::shared_lock<std::shared_mutex> lock_shared(io_mgr_mtx);
    auto iter = io_ctxs.find(fd);
    lock_shared.unlock();
    
    std::unique_lock<std::shared_mutex> lock_unique(io_mgr_mtx);
    if (iter == io_ctxs.end())
    {
        io_ctx = new IOContext;
        io_ctx->fd = fd;
        io_ctxs.insert({fd, io_ctx});
    }
    else {
        io_ctx = iter->second;
    }

    // 将事件添加到epoll中
    epoll_event ep_ev;
    memset(&ep_ev, 0, sizeof(epoll_event));
    int ep_op = (io_ctx->events == (io_ev_t)IOEvent::NONE) ? EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    ep_ev.events = (std::uint32_t)ev;
    ep_ev.data.fd = io_ctx->fd;

    int rt = epoll_ctl(epfd, ep_op, io_ctx->fd, &ep_ev);
    if (rt < 0) {
        warn("epoll_ctl epfd(%d) io_fd(%d) error(%d): %s\n", epfd, io_ctx->fd, errno, strerror(errno));
        return -1;
        EBADF;
    }

    // 注册事件对应的回调函数
    // 边沿触发，主要是为了后面的hook
    io_ctx->events = (std::uint32_t)(io_ctx->events) | (std::uint32_t)ev | EPOLLET;
    ++pending_ev_count;
    auto &ev_ctx = io_ctx->get_ctx(ev);

    // 如果当前线程没有调度器就直接将事件注册到当前的iomanager从属的scheduler中
    if (Scheduler::GetThis() == nullptr) {
        ev_ctx.scher = dynamic_cast<Scheduler *>(this);
    } else {
        ev_ctx.scher = Scheduler::GetThis();
    }

    if (cb) {
        ev_ctx.cb.swap(cb);
    }
    else {
        // 协程将自己进行调度，在进行hook的时候可能会用到
        ev_ctx.fiber = Fiber::GetThis();
        assert(ev_ctx.fiber->get_state() == FiberState::EXEC);
    }
    return 0;
}

bool IOManager::del_ev(int fd, IOEvent ev) {
    if (stopping()) {
      warn("io manager is stopping!\n");
      return -1;
    }

    std::shared_lock<std::shared_mutex> lock_shared(io_mgr_mtx);
    auto iter = io_ctxs.find(fd);
    lock_shared.unlock();

    if (iter == io_ctxs.end()) {
        warn("cannot find fd(%d) in IOManger\n", fd);
        return false;
    }

    std::unique_lock<std::shared_mutex> lock_unique(io_mgr_mtx);
    IOContext *io_ctx = iter->second;

    epoll_event epoll_ev;
    std::uint32_t op = (~(std::uint32_t)ev & (std::uint32_t)io_ctx->events) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_ev.events = ~(std::uint32_t)ev & (std::uint32_t)io_ctx->events;
    epoll_ev.data.fd = io_ctx->fd;

    int rt = epoll_ctl(epfd, op, io_ctx->fd, &epoll_ev);
    if (rt < 0) {
        warn("epoll_ctl(epfd:%d) fail(%d): %s\n", epfd, errno, strerror(errno));
        return false;
    }

    --pending_ev_count;
    io_ctx->events = epoll_ev.events;
    auto &ev_ctx = io_ctx->get_ctx(ev);
    io_ctx->reset_ctx(ev_ctx);
    return true;
}

bool IOManager::cancel_ev(int fd, IOEvent ev) {
    if (stopping()) {
      warn("io manager is stopping!\n");
      return -1;
    }

    std::shared_lock<std::shared_mutex> lock_shared(io_mgr_mtx);
    auto iter = io_ctxs.find(fd);
    lock_shared.unlock();

    if (iter == io_ctxs.end()) {
        warn("cannot find fd(%d) in IOManger\n", fd);
        return false;
    }

    std::unique_lock<std::shared_mutex> lock_unique(io_mgr_mtx);
    IOContext *io_ctx = iter->second;

    epoll_event epoll_ev;
    std::uint32_t op = (~(std::uint32_t)ev & (std::uint32_t)io_ctx->events) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_ev.events = ~(std::uint32_t)ev & (std::uint32_t)io_ctx->events;
    epoll_ev.data.fd = io_ctx->fd;

    int rt = epoll_ctl(epfd, op, io_ctx->fd, &epoll_ev);
    if (rt < 0) {
        warn("epoll_ctl(epfd:%d) fail(%d): %s\n", epfd, errno, strerror(errno));
        return false;
    }

    --pending_ev_count;
    // 触发事件
    io_ctx->trigger_ev(ev);
    return true;
}

bool IOManager::cancel_all(int fd) {
    if (stopping()) {
      warn("io manager is stopping!\n");
      return -1;
    }
  
    std::shared_lock<std::shared_mutex> lock_shared(io_mgr_mtx);
    auto iter = io_ctxs.find(fd);
    lock_shared.unlock();

    if (iter == io_ctxs.end()) {
        warn("cannot find fd(%d)\n", fd);
        return false;
    }

    IOContext *io_ctx = iter->second;
    std::unique_lock<std::shared_mutex> lock_unique(io_mgr_mtx);

    epoll_event epoll_ev;
    std::uint32_t op = EPOLL_CTL_DEL;
    epoll_ev.data.fd = io_ctx->fd;
    epoll_ev.events = (std::uint32_t)io_ctx->events;

    int rt = epoll_ctl(epfd, op, io_ctx->fd, &epoll_ev);
    if (rt < 0) {
        warn("epoll_ctl(epfd:%d fd:%d) fail(%d): %s\n", epfd, io_ctx->fd, errno, strerror(errno));
        return false;
    }

    if ((std::uint32_t)io_ctx->events & (std::uint32_t)IOEvent::READ) {
        --pending_ev_count;
        io_ctx->trigger_ev(IOEvent::READ);
    }
    if ((std::uint32_t)io_ctx->events & (std::uint32_t)IOEvent::WRITE) {
        --pending_ev_count;
        io_ctx->trigger_ev(IOEvent::WRITE);
    }
    if ((std::uint32_t)io_ctx->events & (std::uint32_t)IOEvent::HUP) {
        --pending_ev_count;
        io_ctx->trigger_ev(IOEvent::HUP);
    }
    return true;
}

IOManager* IOManager::GetThis() {
    return dynamic_cast<IOManager*>(Scheduler::GetThis());
}

void IOManager::tickle() {
    static char sign[1] = {'t'};
    chan < sign;
}

bool IOManager::stopping() {
    std::uint64_t timeout = 0;
    return stopping(timeout);
}

void IOManager::on_timer_insert_at_front() {
    tickle();
}

bool IOManager::stopping(std::uint64_t& timeout) {
    timeout = get_next_time();
    return timeout == ~0ull &&
           pending_ev_count == 0 &&
           Scheduler::stopping();
}

void IOManager::idle() {
    debug("idle(%d)\n", Fiber::GetThis()->get_id());
    static const uint16_t EPOLL_MAX = 512;
    epoll_event *evs = new epoll_event[EPOLL_MAX];

    // 构建智能指针并创建析构函数
    std::shared_ptr<epoll_event> events(evs, [](epoll_event *ev){
        delete []ev;
    });

    // 开始调度
    std::uint64_t timeout = 0;
    int num = 0;
    while (!stopping())
    {
        if (pending_ev_count == 0) {
            static thread_local char buf[1];
            chan > buf;
        }

        static const std::uint64_t TIMEOUT_MAX = 3000;
        while (true)
        {
            if (timeout == ~0ull) {
                timeout = TIMEOUT_MAX;
            } else {
                timeout = timeout < TIMEOUT_MAX ? timeout : TIMEOUT_MAX;
            }

            num = epoll_wait(epfd, evs, EPOLL_MAX, timeout);
            if (num < 0 && errno == EINTR) {
                // epoll_wait由于信号，被迫中断了，重新在执行一遍
            } else {
                break;
            }
        }

        // 调度定时器
        std::vector<std::function<void()>> cbs;
        list_expire_cb(cbs);    // 获取被触发的定时器
        if (!cbs.empty()) {
            schedule(cbs.begin(), cbs.end());   // 批量调度
            cbs.clear();
        }
        // 重新设置epoll_wait的超时时间
        timeout = get_next_time();
        
        // 调度io
        for (int i = 0; i < num; ++ i) {

            auto &ev = evs[i];

            std::shared_lock<std::shared_mutex> shared_lock(io_mgr_mtx);
            auto iter = io_ctxs.find(ev.data.fd);
            shared_lock.unlock();

            assert(iter != io_ctxs.end());
            auto io_ctx = iter->second;

            std::unique_lock<std::mutex> lock(io_ctx->mtx);
            if (ev.events & (EPOLLERR | EPOLLHUP)) {
                // 连接出现异常
                auto &ev_ctx = io_ctx->get_ctx(IOEvent::HUP);
                ev_ctx.scher->schedule(ev_ctx.cb);
                continue;
            }

            std::uint32_t real_ev = 0;
            if (ev.events & EPOLLIN) {
                real_ev |= (std::uint32_t)IOEvent::READ;
            }
            if (ev.events & EPOLLOUT) {
                real_ev |= (std::uint32_t)IOEvent::WRITE;
            }
            
            if (!(real_ev & (std::uint32_t)io_ctx->events)) {
                continue;
            }

            // 将触发的事件从epoll_wait中删除
            std::uint32_t del_ev = ~real_ev & (std::uint32_t)io_ctx->events;
            int op = del_ev ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            ev.events = del_ev;
            epoll_ctl(epfd, op, io_ctx->fd, &ev);

            io_ctx->events = del_ev;
            --pending_ev_count;

            if (real_ev & (std::uint32_t)IOEvent::WRITE) {
                auto &ev_ctx = io_ctx->get_ctx(IOEvent::WRITE);
                ev_ctx.scher->schedule(ev_ctx.cb);
            }
            if (real_ev & (std::uint32_t)IOEvent::READ) {
                auto & ev_ctx = io_ctx->get_ctx(IOEvent::READ);
                ev_ctx.scher->schedule(ev_ctx.cb);
            }
        }

        // 挂起
        Fiber::Yield();
    }
}
